package com.afdor.rws.event.driven.channel.producer;

import com.afdor.rws.event.driven.Serialization;
import com.afdor.rws.event.driven.config.EventProducerConfig;
import lombok.Getter;
import lombok.NonNull;
import org.apache.commons.lang3.StringUtils;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

/**
 * 生存者抽象实现
 *
 * @author 悭梵
 * @date Created in 2018-08-15 10:52
 */
public abstract class AbstractEventProducer implements EventProducer {
    /**
     * 初始化状态
     */
    private volatile boolean init = Boolean.FALSE;
    /**
     * 配置信息
     */
    @NonNull
    private EventProducerConfig config;
    /**
     * 对应事件源
     */
    @Getter
    @NonNull
    private String eventSourceName;
    /**
     * 序列化方式
     */
    @Getter
    @NonNull
    private Serialization serialization;

    @Override
    public <T extends EventProducer> T init() {
        if (getConfig() == null) {
            throw new RuntimeException("生产者配置信息不能为空");
        }
        if (getSerialization() == null) {
            throw new RuntimeException("生产者对应事件序列化处理器不能为空");
        }
        if (StringUtils.isBlank(getEventSourceName())) {
            throw new RuntimeException("生产者对应处理事件源不能为空");
        }
        init = Boolean.TRUE;
        return (T) this;
    }

    @Override
    public void destroy() {
        init = Boolean.FALSE;
    }

    @Override
    public <C extends EventProducerConfig> C getConfig() {
        return (C) this.config;
    }

    @Override
    public <T extends EventProducer, C extends EventProducerConfig> T setConfig(@NonNull C config) {
        this.config = config;
        return (T) this;
    }

    @Override
    public <T extends EventProducer> T setEventSourceName(@NonNull String eventSourceName) {
        this.eventSourceName = eventSourceName;
        return (T) this;
    }

    @Override
    public <T extends EventProducer> T setSerialization(@NonNull Serialization serialization) {
        this.serialization = serialization;
        return (T) this;
    }

    /**
     * 验证初始话是否成功
     */
    protected void checkup() {
        int waitCount = 0;
        while (!this.init || waitCount == 100) {
            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));
            waitCount++;
        }
        if (!this.init) {
            throw new IllegalStateException("please execute the init method first.");
        }
    }
}
